// Copyright (c) [2019] [huazhonghua]
// [理解] is licensed under the Mulan PSL v1.
// You can use this software according to the terms and conditions of the Mulan PSL v1.
// You may obtain a copy of Mulan PSL v1 at:
//     http://license.coscl.org.cn/MulanPSL
// THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
// PURPOSE.
// See the Mulan PSL v1 for more details.

#ifndef WORKER_H
#define WORKER_H

#include <pthread.h>
#include <stdio.h>

#include "structure.h"
#include "basic.h"
#include "execution.h"

// queue status
#define ACTIVE 1
#define STOPPING 2

Node* queue_work(Node* all, Node* request) {
	Node* queue = get_linked(all, WORK_QUEUE);
	Node* lock = get_linked(queue, LOCK);
	pthread_mutex_t* mutex = lock->other.pointer;

	pthread_mutex_lock(mutex);

	Node* work = create_node(all);
	work->other.pointer = request;
	Node* queue_tail = get_linked(queue, TAIL);
	if(queue_tail == NULL) {
		link_node(queue, HEAD, work);
		link_node(queue, TAIL, work);
	}
	else {
		link_node(queue_tail, NEXT, work);
		link_node(queue, TAIL, work);
	}

	Node* wait_count = get_linked(queue, WAIT_COUNT);
	if(wait_count->part > 0) {
		Node* cond_holder = get_linked(queue, WORKER_COND);
		pthread_cond_t* cond = cond_holder->other.pointer;
		pthread_cond_signal(cond);
	}

	pthread_mutex_unlock(mutex);

	return NULL;
}

void* worker(void* arg) {
	Node* worker_data = arg;
	Node* all = get_linked(worker_data, ALL);

	Node* i = get_linked(worker_data, I);

/*char name[20];
sprintf(name, "%d", i->part);
FILE* log = fopen(name, "w");
Node* log_node = create_node(all);
log_node->other.pointer = log;*/

	free_node(worker_data);

	Node* queue = get_linked(all, WORK_QUEUE);

	Node* lock = get_linked(queue, LOCK);
	pthread_mutex_t* mutex = lock->other.pointer;

	Node* cond_holder = get_linked(queue, WORKER_COND);
	pthread_cond_t* cond = cond_holder->other.pointer;

	Node* status = get_linked(queue, STATUS);

	Node* wait_count = get_linked(queue, WAIT_COUNT);

	while(1) {
		pthread_mutex_lock(mutex);

		Node* head;
		while(status->part != STOPPING && (head = get_linked(queue, HEAD)) == NULL) {
			++(wait_count->part);
			pthread_cond_wait(cond, mutex);
			--(wait_count->part);
		}

		if(status->part == STOPPING && head == NULL) {
			pthread_mutex_unlock(mutex);
			break;
		}
	
		Node* request = head->other.pointer;

		Node* next = get_linked(head, NEXT);
		if(next == NULL) {
			unlink_node(queue, HEAD);
			unlink_node(queue, TAIL);
		}
		else {
			link_node(queue, HEAD, next);
		}

		free_node(head);
	
		pthread_mutex_unlock(mutex);

//link_node(request, LOG, log_node);

		if(run(request) == REQUEUE) {
			queue_work(all, request);
		}
	}

//	fclose(log);
//	free_node(log_node);

	return NULL;
}

void create_lock(Node* all, Node* node) {
	Node* lock = create_node(all);
	lock->other.pointer = malloc(sizeof(pthread_mutex_t));
	if(!pthread_mutex_init(lock->other.pointer, NULL)) {
	}
	link_node(node, LOCK, lock);
}

void delete_lock(Node* node) {
	Node* lock = get_linked(node, LOCK);
	unlink_node(node, LOCK);
	pthread_mutex_t* mutex = lock->other.pointer;
	pthread_mutex_destroy(mutex);
	free(mutex);
	free_node(lock);
}

void setup_workers(Node* all, int count) {
	Node* queue = create_node(all);
	link_node(all, WORK_QUEUE, queue);

	Node* status = create_node(all);
	status->part = ACTIVE;
	link_node(queue, STATUS, status);

	Node* thread_ids = create_list_node(all, count);
	link_node(queue, CHILD_THREAD_ID, thread_ids);

	create_lock(all, queue);

	Node* cond_holder = create_node(all);
	cond_holder->other.pointer = malloc(sizeof(pthread_cond_t));
	if(!pthread_cond_init(cond_holder->other.pointer, NULL)) {
	}
	link_node(queue, WORKER_COND, cond_holder);

	Node* wait_count = create_node(all);
	wait_count->part = 0;
	link_node(queue, WAIT_COUNT, wait_count);

	for(int i = 0; i < count; ++i) {
		Node* worker_data = create_node(all);

		Node* id_holder = create_node(all);
		id_holder->other.pointer = malloc(sizeof(pthread_t));
		link_node(worker_data, THREAD_ID, id_holder);

		add_to_list_node(thread_ids, id_holder);

		link_node(worker_data, ALL, all);
		link_node(worker_data, I, get_create_int(all, i));

		pthread_create(id_holder->other.pointer, NULL, worker, worker_data);
	}
}

void stop_thread_by_id(Node* id_node) {
	pthread_t* thread_id = id_node->other.pointer; 
	pthread_join(*thread_id, NULL);
	free(thread_id);
	free_node(id_node);
}

void teardown_workers(Node* all) {
	Node* queue = get_linked(all, WORK_QUEUE);

	Node* lock = get_linked(queue, LOCK);
	pthread_mutex_t* mutex = lock->other.pointer;

	pthread_mutex_lock(mutex);

	Node* status = get_linked(queue, STATUS);
	status->part = STOPPING;

	pthread_mutex_unlock(mutex);

	Node* cond_holder = get_linked(queue, WORKER_COND);
	pthread_cond_t* cond = cond_holder->other.pointer;
	pthread_cond_broadcast(cond);

	Node* thread_ids = get_linked(queue, CHILD_THREAD_ID);
	int thread_count = get_list_node_length(thread_ids);
	for(int i = 0; i < thread_count; ++i) {
		Node* id_node = get_list_node_item(thread_ids, i);
		stop_thread_by_id(id_node);
	}
	free_list_node(thread_ids);

	free_node(status);

	Node* wait_count = get_linked(queue, WAIT_COUNT);
	free_node(wait_count);

	pthread_cond_destroy(cond);
	free(cond);
	free_node(cond_holder);

	delete_lock(queue);

	free_node(queue);
}

#endif
